/*
 * Copyright (c) 2021 Huawei Device Co., Ltd.
 * Licensed under the Apache License, Version 2.0 (the 'License');
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an 'AS IS' BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import { RealConnection } from './RealConnection';
import { ExchangeCodec } from '../http/ExchangeCodec';
import { RealCall } from './RealCall';
import { EventListener } from '../EventListener'
import { ExchangeFinder } from './ExchangeFinder';
import { Request } from '../Request'
import { Response } from '../Response'
import { Headers } from '../Headers';
import { RealResponseBody } from '../http/RealResponseBody';
import { ResponseBody } from '../ResponseBody';
import { Sink } from '../okio/Sink';
import { ForwardingSink } from '../okio/ForwardingSink';
import { IOException } from '../Error/error';
import { ProtocolException } from '../Error/error';
import { kotlinCheck } from '../utils/Utils';
import { buffer, util } from '@kit.ArkTS';
import { ForwardingSource } from '../okio/ForwardingSource';
import { Source } from '../okio/Source';
import { Call } from '../Call';
import { ChunkedSource, FixedLengthSource } from '../http1/Http1ExchangeCodec';
import { Gunzip, gunzipSync } from 'fflate';

export class Exchange {
    call: RealCall
    eventListener: EventListener
    finder: ExchangeFinder
    codec: ExchangeCodec
    constructor(call: RealCall, eventListener: EventListener, finder: ExchangeFinder, codec: ExchangeCodec) {
        this.call = call;
        this.eventListener = eventListener;
        this.finder = finder;
        this.codec = codec;
        this.connection = this.codec.connection
    }

    _isDuplex: boolean = false
    private set isDuplex(value: boolean) {
        this._isDuplex = value
    }
    connection: RealConnection



    private _isCoalescedConnection: boolean
    get isCoalescedConnection(): boolean {
        this._isCoalescedConnection = this.finder.address.url.host !== this.connection.route().address.url.host
        return this._isCoalescedConnection
    }

    // @Throws(IOException::class)
    writeRequestHeaders(request: Request) {
        try {
            this.eventListener.requestHeadersStart(this.call)
            this.codec.writeRequestHeaders(request)
            this.eventListener.requestHeadersEnd(this.call, request)
        } catch (e) {
            this.eventListener.requestFailed(this.call, e)
            this.trackFailure(e)
            throw e
        }
    }

    // @Throws(IOException::class)
    async createRequestBody(request: Request, duplex: boolean): Promise<Sink> {
        this._isDuplex = duplex
        let contentLength = await request.body.contentLength()
        this.eventListener.requestBodyStart(this.call)
        let rawRequestBody = this.codec.createRequestBody(request, contentLength)
        return new RequestBodySink(rawRequestBody, contentLength, this)
    }

    // @Throws(IOException::class)
    flushRequest() {
        try {
            this.codec.flushRequest()
        } catch (e) {
            this.eventListener.requestFailed(this.call, e)
            this.trackFailure(e)
            throw e
        }
    }

    // @Throws(IOException::class)
    async finishRequest() {
        try {
            await this.codec.finishRequest()
        } catch (e) {
            this.eventListener.requestFailed(this.call, e)
            this.trackFailure(e)
            throw e
        }
    }

    responseHeadersStart() {
        this.eventListener.responseHeadersStart(this.call)
    }

    // @Throws(IOException::class)
    async readResponseHeaders(expectContinue: boolean){ //: Response.Builder | null {
        try {
            let result = await this.codec.readResponseHeaders(expectContinue)
            result?.initExchange(this)
            return result
        } catch (e) {
            this.eventListener.responseFailed(this.call, e)
            this.trackFailure(e)
            throw e
        }
    }

    responseHeadersEnd(response: Response) {
        this.eventListener.responseHeadersEnd(this.call, response)
    }

    // @Throws(IOException::class)
    openResponseBody(response: Response): ResponseBody | null {
        try {
            let contentType = response.header("Content-Type")
            let contentLength = this.codec.reportedContentLength(response)
            let rawSource = this.codec.openResponseBodySource(response)
            let source = new ResponseBodySource(rawSource, contentLength, this.eventListener, this.call, this)
            return new RealResponseBody(contentType, contentLength, source)
        } catch (e) {
            this.eventListener.responseFailed(this.call, e)
            this.trackFailure(e)
            throw e
        }
    }

    // @Throws(IOException::class)
    trailers(): Headers {
        return this.codec.trailers()
    }

    // @Throws(SocketException::class)
    // newWebSocketStreams(): RealWebSocket.Streams {
    //     this.call.timeoutEarlyExit()
    //     return this.codec.connection.newWebSocketStreams(this)
    // }

    webSocketUpgradeFailed() {
        this.bodyComplete(-1, true, true, null)
    }

    noNewExchangesOnConnection() {
        this.codec.connection.noNewExchanges()
    }

    cancel() {
        this.codec.cancel()
    }

    /**
     * Revoke this exchange's access to streams. This is necessary when a follow-up request is
     * required but the preceding exchange hasn't completed yet.
     */
    detachWithViolence() {
        this.codec.cancel()
        this.call.messageDone(this, true, true, null)
    }

    private trackFailure(e: Error) {
        this.finder.trackFailure(e)
        this.codec.connection.trackFailure(this.call, e)
    }

    // <E : IOException?>
    async bodyComplete(bytesRead: number, responseDone: boolean, requestDone: boolean, e: Error): Promise<Error> {
        if (e !== null && e !== undefined) {
            this.trackFailure(e)
        }
        if (requestDone) {
            if (e !== null && e !== undefined) {
                this.eventListener.requestFailed(this.call, e)
            } else {
                this.eventListener.requestBodyEnd(this.call, bytesRead)
            }
        }
        if (responseDone) {
            if (e !== null && e !== undefined) {
                this.eventListener.responseFailed(this.call, e)
            } else {
                this.eventListener.responseBodyEnd(this.call, bytesRead)
            }
        }
        console.log("okhttp test log :: " , "bodyComplete 开始")

        return await this.call.messageDone(this, requestDone, responseDone, e)
    }

    noRequestBody() {
        this.call.messageDone(this, true, false, null)
    }
}

/** A request body that fires events when it completes. */
export class RequestBodySink extends ForwardingSink{
    constructor(
        delegate: Sink,
        private contentLength: number, // -1 表示未知长度
        private exchange: Exchange
    ) {
        super(delegate);
    }

    private completed = false
    private bytesReceived = 0
    private closed = false

    // @Throws(IOException::class)
    override async write(data: Uint8Array | string, byteCount?: number) {
        kotlinCheck(!this.closed, "closed")
        let chunk: Uint8Array;
        if (typeof data === 'string') {
            chunk = new util.TextEncoder().encodeInto(data);
        } else {
            chunk = data;
        }
        const actualByteCount = chunk.byteLength;
        if (this.contentLength !== -1 && this.bytesReceived + actualByteCount > this.contentLength) {
            throw new ProtocolException(`expected ${this.contentLength} bytes but received ${this.bytesReceived + actualByteCount}`)
        }
        try {
            await super.write(chunk, actualByteCount)
            this.bytesReceived += actualByteCount
        } catch (e) {
            if (e instanceof IOException){
                throw this.complete(e)
            }
        }
    }

        // @Throws(IOException::class)
    override async flush() {
        try {
            await super.flush()
        } catch (e) {
            if (e instanceof IOException){
                throw await this.complete(e)
            }
        }
    }

    // @Throws(IOException::class)
    override async close() {
        if (this.closed) return
        this.closed = true
        if (this.contentLength !== -1 && this.bytesReceived !== this.contentLength) {
            throw new ProtocolException("unexpected end of stream")
        }
        try {
            await super.close()
            await this.complete(null)
        } catch (e) {
            if (e instanceof IOException){
                throw await this.complete(e)
            }
        }
    }

    private async complete(e: Error): Promise<Error> {
        if (this.completed) return e
        this.completed = true
        return await this.exchange.bodyComplete(this.bytesReceived,false,true,e)
    }
}

export class ResponseBodySource extends ForwardingSource {
    private bytesReceived = 0;
    private invokeStartEvent = true;
    private completed = false;
    private closed = false;

    constructor(
        delegate: Source,
        private contentLength: number, // -1 表示未知长度
        private eventListener: EventListener,
        private call: Call,
        private exchange: Exchange
    ) {
        super(delegate);
        // 如果 contentLength 为 0，立即完成
        if (contentLength === 0) {
            this.complete(null);
        }
    }

    override async readString(): Promise<string> {
        if (this.delegate instanceof FixedLengthSource){
            const buff = new Uint8Array(this.contentLength);
            await this.read(buff,this.contentLength)
            let res = buff
            if (this.delegate._isGzip) {
                res = gunzipSync(buff) as Uint8Array
            }
            return buffer.from(res).toString()
        }else if (this.delegate instanceof ChunkedSource){
            let buff: Array<Uint8Array> = [];
            while (true) {
                let temp_buff = new Uint8Array(8192);
                let readCount = await this.read(temp_buff, 8192) // 默认每次最多读 8192 字节
                if (readCount == -1) break
                buff.push(temp_buff.subarray(0, readCount))
            }
            let res
            if (this.delegate._isGzip) {
                res = gunzipSync(new Uint8Array(buffer.concat(buff).buffer))
            } else {
                res = buffer.concat(buff)
            }
            return buffer.from(res).toString()
        }else{
            throw new Error("Method not implemented for UnknownLengthSource.")
        }
    }

    override async readBytes(): Promise<Uint8Array> {
        if (this.delegate instanceof FixedLengthSource) {
            const buff = new Uint8Array(this.contentLength);
            await this.read(buff,this.contentLength)
            let res = buff
            if (this.delegate._isGzip) {
                res = gunzipSync(buff) as Uint8Array
            }
            return res
        }else if (this.delegate instanceof ChunkedSource){
            let buff: Array<Uint8Array> = [];
            while (true) {
                let temp_buff = new Uint8Array(8192);
                let readCount = await this.read(temp_buff, 8192) // 默认每次最多读 8192 字节
                if (readCount == -1) break
                buff.push(temp_buff.subarray(0, readCount))
            }
            let res
            if (this.delegate._isGzip) {
                res = gunzipSync(new Uint8Array(buffer.concat(buff).buffer))
            } else {
                res = new Uint8Array(buffer.concat(buff).buffer)
            }
            return res
        }else{
            throw new Error("Method not implemented for UnknownLengthSource.")
        }
    }

    /**
     * 流式读取响应体，支持 gzip 解压，通过回调返回数据块
     */
    override Stream(callback: (chunk: Uint8Array, complete?:  boolean, error?: Error) => void): void {
        const isGzip = (this.delegate as any)._isGzip === true;
        let gzipStream: Gunzip | null = null;

        if (isGzip) {
            gzipStream = new Gunzip();
            // 设置 ondata 回调：每当解压出一块数据，就触发 onData
            gzipStream.ondata = (data)=>{
                if (data.length > 0) {
                    callback(data, false);
                }
            }
        }

        const readChunk = () => {
            const buff = new Uint8Array(8192);
            this.read(buff, 8192)
                .then(readCount => {
                    if (readCount === -1) {
                        // 流结束：如果是 gzip，需 flush 并结束 inflate
                        if (gzipStream) {
                            try {
                                gzipStream.push(new Uint8Array(0), true); // 结束 inflate
                            } catch (e) {
                                callback(null, false, e instanceof Error ? e : new Error(String(e)));
                                return;
                            }
                        }
                        callback(null, true);
                        return;
                    }

                    const chunk = buff.subarray(0, readCount);

                    if (isGzip) {
                        // 注意：这里传 false，表示“还有更多数据”
                        try {
                            gzipStream!.push(chunk, false);
                        } catch (e) {
                            callback(null, false, e instanceof Error ? e : new Error(String(e)));
                            return;
                        }
                    } else {
                        callback(chunk, false);
                    }

                    // 继续读下一块
                    readChunk();
                })
                .catch(err => {
                    callback(null, false, err instanceof Error ? err : new Error(String(err)));
                });
        };

        // 启动读取
        readChunk();
    }

    override async read(sink: Uint8Array, byteCount: number): Promise<number> {
        kotlinCheck(!this.closed, "closed")


        try {
            const read = await super.read(sink, byteCount);

            // 第一次成功读取时，触发 start 事件
            if (this.invokeStartEvent && read > 0) {
                this.invokeStartEvent = false;
                this.eventListener.responseBodyStart(this.call);
            }

            if (read === -1) {
                // 流结束
                await this.complete(null);
                return -1;
            }

            this.bytesReceived += read;

            // 检查是否超出 Content-Length
            if (this.contentLength !== -1 && this.bytesReceived > this.contentLength) {
                const error = new Error(
                    `Protocol error: expected ${this.contentLength} bytes but received ${this.bytesReceived}`
                );
                await this.complete(error);
                throw error;
            }

            // 如果已读完全部内容
            if (this.contentLength !== -1 && this.bytesReceived === this.contentLength) {
                await this.complete(null);
            }

            return read;
        } catch (e: any) {
            await this.complete(e);
            throw e;
        }
    }

    override async close() {
        if (this.closed) return;
        this.closed = true;
        super.close();
        await this.complete(null);
    }

    /**
     * 标记完成，防止重复触发事件
     */
    private async complete(e: Error): Promise<Error> {
        if (this.completed) return e;
        this.completed = true;

        // 即使没开始读，也要触发 start → end（例如空 body）
        if (e === null && this.invokeStartEvent) {
            this.invokeStartEvent = false;
            this.eventListener.responseBodyStart(this.call);
        }
        console.log("okhttp test log :: " , "complete : 消费响应体")
        return await this.exchange.bodyComplete(this.bytesReceived,true,false,e)
    }
}